package cn.edu.cuit.elena.transaction.controller;

import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

import cn.edu.cuit.elena.common.Result;
import cn.edu.cuit.elena.db.DataSource;
import cn.edu.cuit.elena.nati.shell.SqoopShell;
import cn.edu.cuit.elena.trace.Tracer;
import cn.edu.cuit.elena.transaction.DataSourceManager;
import cn.edu.cuit.elena.transaction.system.UserContext;

/**
 * Sqoop maintains a thread pool, it provides threads to clients, those threads are used to start a
 * UNIXProcess which executes the "sqoop import command".
 * */
public class Sqoop
{
    private static ThreadPoolExecutor server;
    private static Map<UserContext, Sqoop> controllers;

    private UserContext userContext;
    private TaskMonitor taskMonitor;

    static
    {
        server = (ThreadPoolExecutor) Executors.newCachedThreadPool( Executors.defaultThreadFactory() );
        controllers = new HashMap<UserContext, Sqoop>();
    }

    private Sqoop(UserContext userContext, TaskMonitor taskMonitor)
    {
        this.userContext = userContext;
        this.taskMonitor = taskMonitor;
    }

    /**
     * Get Sqoop object for each user and a specified TaskMonitor. One sqoop object subjects to one
     * user, additionally, each user must owns one sqoop object.
     * */
    public static Sqoop get( UserContext userContext, TaskMonitor taskMonitor )
    {
        synchronized( controllers )
        {
            if( controllers.get( userContext ) == null )
            {
                Sqoop sqoop = new Sqoop( userContext, taskMonitor );
                controllers.put( userContext, sqoop );
            }
        }

        return controllers.get( userContext );
    }

    public UserContext getUserContext()
    {
        return userContext;
    }

    /**
     * Import a table of specified datasource, this method will be blocking until importing over or
     * import failed. <strong> The taskId<strong> = dataSourceIdentifier + tableName" as ID.
     * 
     * @return <code>true</code> There is no any exception or error during importing.
     *         <p>
     *         <code>false</code> Any exception or error was thrown from the importing thread.
     * */
    public void doImport( final String dataSourceIdentifier, final String tableName )
    {
        IdentifiedRunnable task = new IdentifiedRunnableImp( dataSourceIdentifier, tableName );
        server.submit( task );
        taskMonitor.monitoringTask( task );
    }

    final class IdentifiedRunnableImp
        implements IdentifiedRunnable
    {

        volatile boolean success = false;
        volatile int progression;

        String dataSourceIdentifier;
        String tableName;
        SqoopShell sqoopShell;
        String taskId;

        IdentifiedRunnableImp(String dataSourceIdentifier, String tableName)
        {
            this.dataSourceIdentifier = dataSourceIdentifier;
            this.tableName = tableName;
            taskId = dataSourceIdentifier + tableName;
        }

        /**
         * Here, return "dataSourceIdentifier + tableName" as ID.
         * */
        @Override
        public String getIdentifier()
        {
            return taskId;
        }

        @Override
        public void run()
        {
            DataSourceManager dataSourceManager = null;

            try
            {
                //Notify monitoring thread that current thread is running.
                synchronized( this )
                {
                    notifyAll();
                }

                synchronized( Sqoop.this.userContext )
                {
                    dataSourceManager = Sqoop.this.userContext.getDataSourceManager();
                }

                for( DataSource dataSource : dataSourceManager.getDataSources() )
                {
                    if( dataSource.toString().equalsIgnoreCase( dataSourceIdentifier ) )
                    {
                        Tracer.debugTrace( Sqoop.IdentifiedRunnableImp.class, "import task started ", null );

                        sqoopShell = new SqoopShell( dataSource, userContext );
                        sqoopShell.importTableToHDFS( tableName );

                        /**
                         * Here! create hive table and associate with import data.
                         * */
                        Hive hive = Hive.get( userContext, dataSource );
                        try
                        {
                            hive.doCreate( tableName );
                            //                            hive.loadDataIntoHiveTable( tableName );
                        }
                        catch( SQLException e )
                        {
                            Tracer.fatalTrace( IdentifiedRunnableImp.class,
                                "Can create hive table and load data due to : " + e.getMessage(), e );
                            progression = -1;
                            return;
                        }

                        Tracer.debugTrace( Sqoop.IdentifiedRunnableImp.class, "import task completed ", null );
                    }
                }

                success = true;
                progression = 100;
            }
            catch( Exception e )
            {
                Tracer.fatalTrace( IdentifiedRunnableImp.class,
                    "IdentifiedRunnableImp run failed due to " + e.getMessage(), e );
                progression = -1;
                success = false;
            }
        }

        /**
         * <strong>Don't use this method.</strong>
         * */
        @Override
        public int getProgression()
        {
            synchronized( this )
            {
                //TODO
                //get current progression           
                return progression;
            }
        }

        @Override
        public String[] getOutput()
        {
            // TODO Auto-generated method stub
            if( sqoopShell == null )
            {
                return null;
            }
            return sqoopShell.getOutput();
        }

        @Override
        public boolean isCompleted()
        {
            return success;
        }

        @Override
        public Result getResult()
        {
            return null;
        }
    }
}
